package ar.uba.is2.common.mensaje.protocolo.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.log4j.Logger;

import ar.uba.is2.common.mensaje.Contenido;
import ar.uba.is2.common.mensaje.ContenidoParticionable;
import ar.uba.is2.common.mensaje.Mensaje;
import ar.uba.is2.common.mensaje.Receptor;
import ar.uba.is2.common.mensaje.TipoMensaje;
import ar.uba.is2.common.mensaje.protocolo.Protocolo;
import ar.uba.is2.common.threads.ThreadManager;

/**
 * Protocolo con Ack y Retry.
 * 
 * Cuando se envía un mensaje:  
 * 
 * - se queda esperando un ACK. Si no llega en
 * {@link #retryTime} ms, entonces reenvío. Si pasados los {@link #timeout} ms
 * no llegó el ACK, el mensaje se descarta.
 * 
 * - parte el mensaje para enviarlo por un canal pequeño.
 * 
 * Cuando se recibe un mensaje:
 * 
 * - se crea un buffer para recibir las distintas partes del mensaje. (mensaje partido)
 * 
 * - cuando se reciben todas las partes, se envía un ack al origen del mensaje.
 * 
 * @author leandro
 */
public class SterenProtocol extends ProtocoloAbstracto {

	private static final Logger log = Logger.getLogger(Protocolo.class);

	private Map<IdentificadorMensaje, Buffer> buffers;
	private Receptor receptor;
	
	private Set<Integer> acksRecibidos;
	private Set<IdentificadorMensaje> acksEnviados;
	
	private long retryTime;
	private long timeout;

	public SterenProtocol(Receptor receptor, long retryTime, long timeout) {
		this.buffers = new HashMap<IdentificadorMensaje, Buffer>();
		this.acksRecibidos = new HashSet<Integer>();
		this.acksEnviados = new HashSet<IdentificadorMensaje>();
		this.receptor = receptor;
		this.retryTime = retryTime;
		this.timeout = timeout;
	}

	public void enviarMensaje(Mensaje msj) {
		log.info(String.format("Enviar mensaje %s con destino %s", msj.getNumeroMensaje(), msj.getDestino()));

		EnviarMensajeDaemon daemon = new EnviarMensajeDaemon(msj, retryTime, timeout);
		ThreadManager.instance.crearYCorrerThread(daemon);
	}

	public void recibirMensaje(Mensaje mensaje) {
		switch (mensaje.getTipoMensaje()) {
		case ACK:
			recibirMensajeAck(mensaje);
			break;
		case DATOS_SENSADOS:
			recibirMensajeDatos(mensaje);
			break;
		}
	}

	private boolean ackEnv(Mensaje msj) {
		return this.acksEnviados.contains(crearIdMensajeDatos(msj));
	}
	private boolean ackRec(Mensaje msj) {
		return this.acksRecibidos.contains(crearIdMensajeAck(msj));
	}

	private void avisoDescartarMensaje(Mensaje msj, String causa) {
		log.warn("Descartando mensaje " + msj.getNumeroMensaje() + "\nCausa: " + causa);
	}

	private Buffer crearBuffer(Mensaje msj) {
		Buffer buff = new Buffer(msj.getNumeroMensaje(), msj.getCantidadPartes(), msj.getOrigen(), msj.getDestino(), msj.getFechaCreacion(), msj.getTipoMensaje());

		return buff;
	}


	private Integer crearIdMensajeAck(Mensaje msj) {
		return msj.getNumeroMensaje();

	}

	private IdentificadorMensaje crearIdMensajeDatos(Mensaje mensaje) {
		return new IdentificadorMensaje(mensaje.getNumeroMensaje(), mensaje.getOrigen().getNombre());
	}

	private void despacharMensaje(Mensaje msjReagrupado) {
		this.receptor.recibirMensaje(msjReagrupado);

	}

	private void enviarAck(Mensaje msjReagrupado) {

		Mensaje mensaje = new Mensaje();

		mensaje.setNumeroMensaje(msjReagrupado.getNumeroMensaje());
		mensaje.setTipoMensaje(TipoMensaje.ACK);
		mensaje.setOrigen(msjReagrupado.getDestino());
		mensaje.setDestino(msjReagrupado.getOrigen());

		this.canal.enviarMensaje(mensaje);
		this.acksEnviados.add(crearIdMensajeDatos(msjReagrupado));
	}

	private void recibirMensajeAck(Mensaje mensaje) {
		log.info(String.format("Mensaje ACK %s recibido", mensaje.getNumeroMensaje()));
		this.acksRecibidos.add(crearIdMensajeAck(mensaje));
	}
	
	private void recibirMensajeDatos(Mensaje mensaje) {
		// si ya envié el ack de este mensaje, entonces es un mensaje repetido
		if (ackEnv(mensaje)) {
			
			avisoDescartarMensaje(mensaje, "Mensaje repetido");
			return;
		}
		
		log.info(String.format("Mensaje recibido %s: (%s/%s)", mensaje.getNumeroMensaje(), mensaje.getParte(), mensaje.getCantidadPartes()));

		IdentificadorMensaje bufferKey = crearIdMensajeDatos(mensaje);
		synchronized (buffers) {
			if (!buffers.containsKey(bufferKey)) {
				Buffer b = crearBuffer(mensaje);
				buffers.put(bufferKey, b);
			}
		}

		Buffer buff = buffers.get(bufferKey);

		buff.append(mensaje);

		if (buff.estaCompleto()) {
			Mensaje msjReagrupado = buff.reagruparMensaje();
			log.info(String.format("Mensaje completo <%s,%s>", msjReagrupado.getOrigen(), msjReagrupado.getNumeroMensaje()));
			enviarAck(msjReagrupado);
			despacharMensaje(msjReagrupado);
		}
	}


	private class EnviarMensajeDaemon implements Runnable {
		
		private final Logger log = Logger.getLogger(EnviarMensajeDaemon.class);


		private Mensaje msj;
		private List<Mensaje> msjsPartidos;
		private long timeout = 120000l;
		private long retryTime = 15000l;
		private long waitingTimeToCheckAck = retryTime / 10;

		private EnviarMensajeDaemon(Mensaje msjAEnviar, long retryTime, long timeout) {
			this.msj = msjAEnviar;
			this.msjsPartidos = partirMensaje();
			this.retryTime = retryTime;
			this.waitingTimeToCheckAck = retryTime / 10;
			this.timeout = timeout;
		}

		public void run() {
			try {
				while (!ackRec(msj) && !timeout(msj)) {
					
					log.info(String.format("Intentando enviar mensaje <%s,%s>", msj.getDestino(), msj.getNumeroMensaje()));
					
					for (Mensaje parteMsj : this.msjsPartidos) {
						
						if ( log.isDebugEnabled() ) {
							log.debug(String.format("Intentando enviar parte (%s/%s): ", parteMsj.getParte(), parteMsj.getCantidadPartes()));
						}
						SterenProtocol.this.canal.enviarMensaje(parteMsj);
					}
					
					synchronized (this) { // sólo para hacer wait()
						int i = 0;
						while(!ackRec(msj) && i < 10){
							this.wait(waitingTimeToCheckAck);
						}
					}
				}
				if (!ackRec(msj)) {
					log.info(String.format("Mensaje %s sin ack. TIMEOUT", msj.getNumeroMensaje()));
					avisoDescartarMensaje(msj, "Timeout");
				} else {
					log.info(String.format("Mensaje %s enviado OK", msj.getNumeroMensaje()));
				}
			} catch (InterruptedException e) {
			}
		}


		/**
		 * Arma la los mensajes que se enviaran a la EC. Estos mensajes respetan el
		 * tamano maximo del canal.
		 * 
		 * @param datosSensadoIt
		 *            datos que se desean enviar.
		 * @return los mensaje sin la cantidad total de partes.
		 */
		private List<Mensaje> partirMensaje() {
			boolean dbgEnabled = log.isDebugEnabled();
			
			if (dbgEnabled){
				log.debug(String.format("Partir mensaje <%s,%s>", msj.getDestino(), msj.getNumeroMensaje()));
			}
			
			Collection<Contenido> contenidos;
			
			Contenido contenido = msj.getContenido();
			if (contenido instanceof ContenidoParticionable) {
				contenidos = ((ContenidoParticionable)contenido).partirContenido(SterenProtocol.this.canal.getTamanioMaximoMensaje());
			} else {
				contenidos = new ArrayList<Contenido>();
				contenidos.add(contenido);
			}
			
			int parte = 0;
			List<Mensaje> mensajesPartidos = new ArrayList<Mensaje>();
			for (Contenido cont : contenidos) {
				Mensaje parteMsj = new Mensaje();
				parteMsj.setDestino(msj.getDestino());
				parteMsj.setOrigen(msj.getOrigen());
				parteMsj.setNumeroMensaje(msj.getNumeroMensaje());
				parteMsj.setFechaCreacion(msj.getFechaCreacion());
				parteMsj.setCantidadPartes(contenidos.size());
				parteMsj.setParte(parte);
				parteMsj.setTipoMensaje(msj.getTipoMensaje());
				parteMsj.setContenido(cont);
				parte++;
				
				if ( dbgEnabled ) {
					log.debug(String.format("Parte %s del mensaje <%s,%s> creada", parteMsj.getParte(), parteMsj.getDestino(), parteMsj.getNumeroMensaje()));
				}
			}
			
			return mensajesPartidos;
		}

		private boolean timeout(Mensaje msj) {
			long now = new Date().getTime();

			long creation = msj.getFechaCreacion().getTime();

			return now - creation > timeout;
		}
	}

	private static class IdentificadorMensaje {
		private Integer nroMensaje;
		private String nombreOrigen;

		public IdentificadorMensaje(Integer nroMensaje, String nombreOrigen) {
			this.nroMensaje = nroMensaje;
			this.nombreOrigen = nombreOrigen;
		}

		@Override
		public boolean equals(Object obj) {
			if (!(obj instanceof IdentificadorMensaje)) {
				return false;
			}
			IdentificadorMensaje that = (IdentificadorMensaje) obj;
			return new EqualsBuilder().append(this.nroMensaje, that.nroMensaje)
					.append(this.nombreOrigen, that.nombreOrigen).isEquals();
		}

		@Override
		public int hashCode() {
			return new HashCodeBuilder().append(this.nroMensaje).append(
					this.nombreOrigen).toHashCode();
		}
	}
	
}
